package cn.sciento.transfer.app.service.impl;

import com.alibaba.fastjson.JSON;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import org.apache.commons.collections4.CollectionUtils;
import cn.sciento.transfer.app.service.ConsumerEventService;
import cn.sciento.transfer.app.service.EventErrorService;
import cn.sciento.transfer.domain.entity.ConsDbConfig;
import cn.sciento.transfer.domain.entity.ConsumerEvent;
import cn.sciento.transfer.domain.entity.ConsumerEventArch;
import cn.sciento.transfer.domain.entity.EventError;
import cn.sciento.transfer.infra.mapper.JdbcConsumerEventMapper;
import cn.sciento.transfer.infra.rocketmq.vo.MqMessage;
import cn.sciento.transfer.infra.rocketmq.vo.SqlInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.RowMapper;

public class ConsumerEventServiceImpl implements ConsumerEventService {
  private static final Logger LOGGER = LoggerFactory.getLogger(ConsumerEventServiceImpl.class);
  
  public static final String CONSUMER_TABLE_NAME = "hdtt_consumer_event";
  
  public static final String CONSUMER_TABLE_NAME_ARCH = "hdtt_consumer_event_arch";
  
  private EventErrorService eventErrorService;
  
  public ConsumerEventServiceImpl(EventErrorService eventErrorService) {
    this.eventErrorService = eventErrorService;
  }
  
  public void saveMessageToDb(ConsDbConfig consDbConfig, List<MqMessage> msgList, JdbcTemplate jdbcTemplate) {
    if (CollectionUtils.isEmpty(msgList))
      return; 
    for (MqMessage mqMessage : msgList) {
      ConsumerEvent consumerEvent = new ConsumerEvent();
      consumerEvent.copyFieldsFromMqMessage(mqMessage);
      consumerEvent.setSourceTenantId(mqMessage.getTenantId());
      consumerEvent.setSourceKeyId(mqMessage.getSourceKeyId());
      consumerEvent.setProcessTime(new Date());
      consumerEvent.setProduceTime(mqMessage.getProduceTime());
      consumerEvent.setProcessStatus("P");
      consumerEvent.setTargetService(consDbConfig.getConsumerService());
      consumerEvent.setTargetDb(consDbConfig.getConsumerDb());
      consumerEvent.setTargetTable(consDbConfig.getConsumerTable());
      consumerEvent.setTargetDsId(consDbConfig.getConsumerDsId());
      consumerEvent.setTargetDs(consDbConfig.getConsumerDs());
      try {
        consumerEvent.setMessages(JSON.toJSONString(mqMessage));
      } catch (Exception e) {
        LOGGER.error(e.getMessage(), e);
      } 
      SqlInfo<String, List<Object>> sqlInfo = SqlInfo.builderInsertSqlFromEntity(consumerEvent, getConsumerTable("hdtt_consumer_event", consDbConfig, jdbcTemplate));
      jdbcTemplate.update(sqlInfo.getSql(), sqlInfo.getData().toArray());
    } 
  }
  
  public void updateMessageStatus(ConsDbConfig consDbConfig, MqMessage mqMessage, String status, String msg, JdbcTemplate jdbcTemplate) {
    ConsumerEvent consumerEvent = new ConsumerEvent();
    consumerEvent.setProcessStatus(status);
    consumerEvent.setProcessMsg(msg);
    consumerEvent.setProcessTime(new Date());
    ConsumerEvent whereInfo = new ConsumerEvent();
    whereInfo.setEventId(mqMessage.getEventId());
    whereInfo.setSourceKeyId(mqMessage.getSourceKeyId());
    whereInfo.setTargetDb(consDbConfig.getConsumerDb());
    whereInfo.setTargetTable(consDbConfig.getConsumerTable());
    SqlInfo<String, List<Object>> sqlInfo = SqlInfo.builderUpdateSqlFromEntity(consumerEvent, whereInfo, getConsumerTable("hdtt_consumer_event", consDbConfig, jdbcTemplate));
    jdbcTemplate.update(sqlInfo.getSql(), sqlInfo.getData().toArray());
  }
  
  public boolean archMessageIfExists(ConsDbConfig consDbConfig, MqMessage mqMessage, JdbcTemplate jdbcTemplate) {
    mqMessage.setTargetDb(consDbConfig.getConsumerDb());
    mqMessage.setTargetTable(consDbConfig.getConsumerTable());
    int consumerCount = selectMsgList(mqMessage, null, getConsumerTable("hdtt_consumer_event", consDbConfig, jdbcTemplate), "=", jdbcTemplate).size();
    if (consumerCount > 0) {
      updateMessageStatus(consDbConfig, mqMessage, "S", "Reapeat Msg", jdbcTemplate);
      moveMqMessageToArchTable(consDbConfig, mqMessage, jdbcTemplate);
      return true;
    } 
    int consumerArchCount = selectMsgList(mqMessage, null, getConsumerTable("hdtt_consumer_event_arch", consDbConfig, jdbcTemplate), "=", jdbcTemplate).size();
    return (consumerArchCount > 0);
  }
  
  public List<MqMessage> loadNMsgListLargeThanEventId(ConsDbConfig consDbConfig, MqMessage mqMessage, JdbcTemplate jdbcTemplate) {
    MqMessage queryMsg = new MqMessage();
    queryMsg.setTargetDb(consDbConfig.getConsumerDb());
    queryMsg.setTargetTable(consDbConfig.getConsumerTable());
    queryMsg.setEventId(mqMessage.getEventId());
    return selectMsgList(queryMsg, "N", getConsumerTable("hdtt_consumer_event", consDbConfig, jdbcTemplate), ">", jdbcTemplate);
  }
  
  public List<MqMessage> loadNMsgListLessThanEventId(ConsDbConfig consDbConfig, MqMessage mqMessage, JdbcTemplate jdbcTemplate) {
    return selectMsgList(mqMessage, "N", getConsumerTable("hdtt_consumer_event", consDbConfig, jdbcTemplate), "<", jdbcTemplate);
  }
  
  public void moveMqMessageToArchTable(ConsDbConfig consDbConfig, MqMessage mqMessage, JdbcTemplate jdbcTemplate) {
    ConsumerEvent whereInfo = new ConsumerEvent();
    whereInfo.setEventId(mqMessage.getEventId());
    whereInfo.setSourceKeyId(mqMessage.getSourceKeyId());
    whereInfo.setTargetDb(consDbConfig.getConsumerDb());
    whereInfo.setTargetTable(consDbConfig.getConsumerTable());
    SqlInfo<String, List<Object>> sqlInfo = SqlInfo.builderSelectAllSqlFromEntity(whereInfo, getConsumerTable("hdtt_consumer_event", consDbConfig, jdbcTemplate), new SqlInfo.KeyValueBind[0]);
    List<ConsumerEvent> consumerEventList = jdbcTemplate.query((String)sqlInfo.getSql(), ((List)sqlInfo
        .getData()).toArray(), (RowMapper)new JdbcConsumerEventMapper());
    if (CollectionUtils.isNotEmpty(consumerEventList)) {
      ConsumerEventArch consumerEventArch = new ConsumerEventArch();
      BeanUtils.copyProperties(consumerEventList.get(0), consumerEventArch);
      SqlInfo<String, List<Object>> archSqlInfo = SqlInfo.builderInsertSqlFromEntity(consumerEventArch, getConsumerTable("hdtt_consumer_event_arch", consDbConfig, jdbcTemplate));
      jdbcTemplate.update((String)archSqlInfo.getSql(), ((List)archSqlInfo.getData()).toArray());
    } 
    SqlInfo<String, List<Object>> deleteSql = SqlInfo.builderDeleteSqlFromEntity(whereInfo, getConsumerTable("hdtt_consumer_event", consDbConfig, jdbcTemplate));
    jdbcTemplate.update((String)deleteSql.getSql(), ((List)deleteSql.getData()).toArray());
  }
  
  public boolean checkMqMessageIsExpired(ConsDbConfig consDbConfig, MqMessage mqMessage, JdbcTemplate jdbcTemplate) {
    return (selectMsgList(mqMessage, "S", getConsumerTable("hdtt_consumer_event_arch", consDbConfig, jdbcTemplate), ">", jdbcTemplate).size() > 0);
  }
  
  public void errorHandler(MqMessage mqMessage, ConsDbConfig consDbConfig, String errorMsg) {
    EventError eventError = new EventError();
    eventError.setEventId(mqMessage.getEventId());
    eventError.setEventType(mqMessage.getType());
    eventError.setSourceService(mqMessage.getSourceService());
    eventError.setSourceTable(mqMessage.getTable());
    eventError.setSourceKeyId(mqMessage.getSourceKeyId());
    eventError.setSourceTenantId(mqMessage.getTenantId());
    eventError.setProduceTime(mqMessage.getProduceTime());
    eventError.setTargetService(consDbConfig.getConsumerService());
    eventError.setTargetDsId(consDbConfig.getConsumerDsId());
    eventError.setTargetDb(consDbConfig.getConsumerDb());
    eventError.setTargetTable(consDbConfig.getConsumerTable());
    eventError.setTargetDs(consDbConfig.getConsumerDs());
    try {
      eventError.setMessages(JSON.toJSONString(mqMessage));
    } catch (Exception e) {
      if (LOGGER.isDebugEnabled())
        LOGGER.debug("error convert MqMessage to JSON"); 
    } 
    eventError.setProcessStatus("P");
    eventError.setErrorType("CONSUME");
    eventError.setErrorText(errorMsg);
    eventError.setErrorTime(new Date());
    this.eventErrorService.asyncInsertEventError(eventError);
  }
  
  public boolean checkMsgInArch(ConsDbConfig consDbConfig, MqMessage mqMessage, JdbcTemplate jdbcTemplate) {
    mqMessage.setTargetDb(consDbConfig.getConsumerDb());
    mqMessage.setTargetTable(consDbConfig.getConsumerTable());
    int consumerArchCount = selectMsgList(mqMessage, null, getConsumerTable("hdtt_consumer_event_arch", consDbConfig, jdbcTemplate), "=", jdbcTemplate).size();
    return (consumerArchCount > 0);
  }
  
  private List<MqMessage> selectMsgList(MqMessage mqMessage, String status, String tableName, String eventIdPattern, JdbcTemplate jdbcTemplate) {
    List<MqMessage> mqMessageList = new ArrayList<>();
    ConsumerEvent queryMsg = new ConsumerEvent();
    queryMsg.setSourceKeyId(mqMessage.getSourceKeyId());
    queryMsg.setProcessStatus(status);
    queryMsg.setSourceService(mqMessage.getSourceService());
    queryMsg.setSourceTable(mqMessage.getTable());
    queryMsg.setSourceTenantId(mqMessage.getTenantId());
    queryMsg.setTargetDb(mqMessage.getTargetDb());
    queryMsg.setTargetTable(mqMessage.getTargetTable());
    SqlInfo<String, List<Object>> sqlInfo = SqlInfo.builderSelectAllSqlFromEntity(queryMsg, tableName, new SqlInfo.KeyValueBind[] { new SqlInfo.KeyValueBind("eventId", mqMessage.getEventId(), eventIdPattern) });
    List<ConsumerEvent> consumerEventList = jdbcTemplate.query((String)sqlInfo.getSql(), (RowMapper)new JdbcConsumerEventMapper(), ((List)sqlInfo.getData()).toArray());
    for (ConsumerEvent consumerEvent : consumerEventList) {
      MqMessage msg = null;
      try {
        msg = (MqMessage)JSON.parseObject(consumerEvent.getMessages(), MqMessage.class);
      } catch (Exception e) {
        if (LOGGER.isDebugEnabled())
          LOGGER.debug("JSON convert to MqMessage Failed", e); 
      } 
      mqMessageList.add(msg);
    } 
    return mqMessageList;
  }
  
  private static String getConsumerTable(String table, ConsDbConfig consDbConfig, JdbcTemplate jdbcTemplate) {
    if (table.equals("hdtt_consumer_event"))
      return consDbConfig.getConsumerEvent(); 
    return consDbConfig.getConsumerEventArch();
  }
}
